package com.zero.kafka;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import kafka.common.OffsetAndMetadata;
import kafka.coordinator.group.BaseKey;
import kafka.coordinator.group.GroupMetadataManager;
import kafka.coordinator.group.OffsetKey;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

public class KafkaCli {

    public static void main(String[] args) throws Exception {
        Consumer<byte[], byte[]> consumer = createKafkaConsumer();
        consumer.subscribe(Lists.newArrayList("__consumer_offsets"));
        while (true) {
            ConsumerRecords<byte[], byte[]> records = consumer.poll(100);
            Iterator<ConsumerRecord<byte[], byte[]>> iterator = records.iterator();
            Map<String, Integer> map = Maps.newHashMap();
            while (iterator.hasNext()) {
                ConsumerRecord<byte[], byte[]> record = iterator.next();
                if (record.key() == null) {
                    continue;
                }
                BaseKey baseKey = GroupMetadataManager.readMessageKey(ByteBuffer.wrap(record.key()));
                byte[] value = record.value();
                if (value == null) {
                    continue;
                }
                OffsetAndMetadata offset = GroupMetadataManager.readOffsetMessageValue(ByteBuffer.wrap(value));
                if (baseKey instanceof OffsetKey) {
                    OffsetKey newKey = (OffsetKey) baseKey;
                    String group = newKey.key().group();
                    TopicPartition tp = newKey.key().topicPartition();
                    System.out.println(group + "," + tp.topic() + "," + tp.partition() + "," + offset.offset());
                }
            }
        }
    }
    static Consumer<byte[], byte[]> createKafkaConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.0.50:9092,192.168.0.50:9093,192.168.0.50:9094");
        props.put("group.id", "KafkaConsumerDemo");
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        return new KafkaConsumer<byte[], byte[]>(props);
    }
}
